-module(rtps_cfr).

-include("rtps.hrl").
-include("rtps_cfr.hrl").

-export([new/0, new/2, new/3,
	 acked_changes_set/2, next_requested_change/1, next_unsent_change/1,
	 requested_changes/1, requested_changes_set/2, unsent_changes/1,
	 unacked_changes/1,
	 unacked_changes_set/2, underway_changes_set/2, irrelevant_change_set/2,
	 is_acked/2]).

-export_type([cfr/0]).

-type cfr() :: #cfr{}.

%% The following functions are implementations of the functions from
%% sections 8.4.7.5.xx in which the logic is copied literally as far
%% as possible.

-spec new() -> Cfrs when
      Cfrs :: [cfr()].
new() ->
    [].

-spec new(Seq_nums, DDS_filter) -> Cfrs when
      Seq_nums :: nonempty_list(rtps:sequence_number()),
      DDS_filter :: fun((rtps:sequence_number()) -> boolean()),
      Cfrs :: nonempty_list(cfr()).
%% 8.4.7.5.1 new
new(Seq_nums, DDS_filter) ->
    [#cfr{sequence_number = N, status = unsent, is_relevant = not DDS_filter(N)}
     || N <- Seq_nums].

-spec new(Min_seq_num, Max_seq_num, DDS_filter) -> Cfrs when
      Min_seq_num :: rtps:sequence_number(),
      Max_seq_num :: rtps:sequence_number(),
      DDS_filter :: fun((rtps:sequence_number()) -> boolean()),
      Cfrs :: nonempty_list(cfr()).
%% 8.4.7.5.1 new
new(Min_seq_num, Max_seq_num, DDS_filter) ->
    new(lists:seq(Min_seq_num, Max_seq_num), DDS_filter).

-spec acked_changes_set(Cfrs1, Commited_seq_num) -> Cfrs2 when
      Cfrs1 :: [cfr()],
      Commited_seq_num :: rtps:sequence_number(),
      Cfrs2 :: [(cfr())].
%% 8.4.7.5.2 acked_changes_set This operation changes the
%% ChangeForReader status of a set of changes for the reader
%% represented by ReaderProxy ‘the_reader_proxy.’ The set of changes
%% with sequence number smaller than or equal to the value
%% ‘committed_seq_num’ have their status changed to ACKNOWLEDGED.
acked_changes_set(Cfrs, Commited_seq_num) ->
    Fun = fun(#cfr{sequence_number = N} = Cfr) 
		when N =< Commited_seq_num ->
		  Cfr#cfr{status = acknowledged};
	     (Cfr) ->
		  Cfr
	  end,
    Cfrs1 = lists:map(Fun, Cfrs),
    Cfrs2 = shrink(Cfrs1),
    Cfrs2.

-spec next_requested_change(Cfrs) -> Cfr when
      Cfrs :: nonempty_list(cfr()),
      Cfr :: cfr().
%% 8.4.10.4.2 available_changes_max This operation returns the maximum
%% SequenceNumber_t among the changes_from_writer changes in the RTPS
%% WriterProxy that are available for access by the DDS DataReader.
next_requested_change(Cfrs) ->
    [Cfr | _] = requested_changes(Cfrs),
    Cfr.

-spec next_unsent_change(Cfrs) -> Cfr when
      Cfrs :: [cfr()],
      Cfr :: cfr().
%% 8.4.7.5.4 next_unsent_change. This operation returns the CacheChange
%% for the ReaderProxy that has the lowest sequence number among the
%% changes with status ‘UNSENT.’ This represents the next change that
%% should be sent to the RTPS Reader represented by the ReaderProxy.
next_unsent_change(Cfrs) ->
    [Cfr | _] = unsent_changes(Cfrs),
    Cfr.

-spec requested_changes(Cfrs1) -> Cfrs2 when
      Cfrs1 :: [cfr()],
      Cfrs2 :: [cfr()].
%% 8.4.7.5.5 requested_changes This operation returns the subset of
%% changes for the ReaderProxy that have status ‘REQUESTED.’ This
%% represents the set of changes that were requested by the RTPS
%% Reader represented by the ReaderProxy using an ACKNACK Message.
requested_changes(Cfrs) ->
    [Cfr || #cfr{status = requested} = Cfr <- Cfrs].

-spec requested_changes_set(Cfrs1, Req_seq_num_set) -> Cfrs2 when
      Cfrs1 :: [cfr()],
      Req_seq_num_set :: [rtps:sequence_number()],
      Cfrs2 :: [cfr()].
%% 8.4.7.5.6 requested_changes_set. This operation modifies the
%% ChangeForReader status of a set of changes for the RTPS Reader
%% represented by ReaderProxy ‘this.’ The set of changes with sequence
%% numbers ‘req_seq_num_set’ have their status changed to REQUESTED.
requested_changes_set(Cfrs, Req_seq_num_set) ->
    changes_set_status(Cfrs, Req_seq_num_set, requested).

-spec unsent_changes(Cfrs1) -> Cfrs2 when
      Cfrs1 :: [cfr()],
      Cfrs2 :: [cfr()].
%% 8.4.7.5.7 unsent_changes. This operation returns the subset of
%% changes for the ReaderProxy the have status ‘UNSENT.’ This
%% represents the set ofchanges that have not been sent to the RTPS
%% Reader represented by the ReaderProxy.
unsent_changes(Cfrs) ->
    [Cfr || #cfr{status = unsent} = Cfr <- Cfrs].

-spec unacked_changes(Cfrs1) -> Cfrs2 when
      Cfrs1 :: [cfr()],
      Cfrs2 :: [cfr()].
%% 8.4.7.5.8 unacked_changes. This operation returns the subset of
%% changes for the ReaderProxy that have status ‘UNACKNOWLEDGED.’ This
%% represents the set of changes that have not been acknowledged yet
%% by the RTPS Reader represented by the ReaderProxy.
unacked_changes(Cfrs) ->
    [Cfr || #cfr{status = unacknowledged} = Cfr <- Cfrs].

%% -- not in specs -----

-spec underway_changes_set(Cfrs1, Req_seq_num_set) -> Cfrs2 when
      Cfrs1 :: [cfr()],
      Req_seq_num_set :: [rtps:sequence_number()],
      Cfrs2 :: [cfr()].
%% This operation is not defined in the specs, but is needed to set
%% the status of cfrs to underway.
underway_changes_set(Cfrs1, Req_seq_num_set) ->
    changes_set_status(Cfrs1, Req_seq_num_set, underway).

-spec unacked_changes_set(Cfrs1, Req_seq_num_set) -> Cfrs2 when
      Cfrs1 :: [cfr()],
      Req_seq_num_set :: [rtps:sequence_number()],
      Cfrs2 :: [cfr()].
%% This operation is not defined in the specs, but is needed to set
%% the status of cfrs to unacknowledged.
unacked_changes_set(Cfrs1, Req_seq_num_set) ->
    changes_set_status(Cfrs1, Req_seq_num_set, unacknowledged).

-spec irrelevant_change_set(Cfrs1, Seq_nums) -> Cfrs2 when
      Cfrs1 :: nonempty_list(cfr()),
      Seq_nums :: [rtps:sequence_number()],
      Cfrs2 :: nonempty_list(cfr()).
%% This operation modifies the status of a ChangeFromWriter to
%% indicate that the CacheChange with the SequenceNumber_t ‘a_seq_num’
%% is irrelevant to the RTPS Writer: i.e. the CacheChange is removed.
irrelevant_change_set(Cfrs, []) ->
    Cfrs;
irrelevant_change_set(Cfrs, [Seq_num | Seq_nums]) ->
    Cfrs1 = case lists:keyfind(Seq_num, #cfr.sequence_number, Cfrs) of
		false ->
		    Cfrs;
		Cfr ->
		    %% Cfr1 = Cfr#cfr{status = acknowledged, is_relevant = false},
		    Cfr1 = Cfr#cfr{is_relevant = false},
		    lists:keyreplace(Seq_num, #cfr.sequence_number, Cfrs, Cfr1)
	    end,
    irrelevant_change_set(Cfrs1, Seq_nums).

-spec is_acked(Cfrs, Seq_num) -> boolean() when
      Cfrs :: nonempty_list(cfr()),
      Seq_num :: rtps:sequence_number().
%% @doc Returns true if the cache change has been been acknowledged by the reader.      
is_acked([#cfr{sequence_number = N} | _], Seq_num) 
  when Seq_num < N ->
    true;
is_acked(Cfrs, Seq_num) ->
    case lists:keyfind(Seq_num, #cfr.sequence_number, Cfrs) of
	#cfr{status = acknowledged, is_relevant = true} ->
	    true;
	false ->
	    false
    end.

    

%%%===================================================================
%%% Internal functions
%%%===================================================================

%% Shrink the cfrs if possible by removing cfrs with status
%% acknowledged from the beginning of the list.
shrink([#cfr{status = acknowledged} |Cfrs]) ->
    shrink(Cfrs);
shrink(Cfrs) ->
    Cfrs.

%% Set the status of the cfrs given there sequence numbers
changes_set_status(Cfrs1, Req_seq_num_set, Status) ->
    Fun = fun(#cfr{sequence_number = N} = Cfr, [N | Ns]) ->
		  {Cfr#cfr{status = Status}, Ns};
	     (Cfr, Ns) ->
		  {Cfr, Ns}
	  end,
    {Cfrs2, _} = lists:mapfoldl(Fun, Req_seq_num_set, Cfrs1),
    Cfrs2.
